package com.zhm.demo.ws;

import com.zhm.demo.ws.simple.MyHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.web.socket.config.annotation.*;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

import java.util.Map;
import java.util.concurrent.TimeUnit;

@Slf4j
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfiguration implements WebSocketMessageBrokerConfigurer, WebSocketConfigurer {

    @Autowired
    @Lazy
    private SimpMessagingTemplate messagingTemplate;


    @Lazy
    @Autowired
    private MyHandler myHandler;

    @Bean
    public ThreadPoolTaskScheduler taskScheduler() {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setPoolSize(1);  // 设置线程池大小
        taskScheduler.setThreadNamePrefix("wss-heartbeat-thread-");  // 设置线程名前缀
        taskScheduler.initialize();
        return taskScheduler;
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        //客户端向服务端发起请求时，需要以前缀。
//        registry.setApplicationDestinationPrefixes("");
        //给指定用户发送一对一的消息前缀是。
        registry.setUserDestinationPrefix("/client");
        // 配置消息代理，这里使用内置的简单消息代理，也可以使用其他消息代理如RabbitMQ
        registry.enableSimpleBroker()
                .setHeartbeatValue(new long[]{50000, 20000}) // 设置心跳间隔（发送、接收），单位是毫秒
                .setTaskScheduler(taskScheduler());
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry
                .addEndpoint("/push/ws/stomp")
                .addInterceptors(myHandshake())
                .setAllowedOriginPatterns("*");
        registry
                .addEndpoint("/push/sockjs/stomp")
                .addInterceptors(myHandshake())
                .setAllowedOriginPatterns("*")
                .withSockJS();//为/marcopolo路径启用SockJS功能
    }

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        // 注册websocket实现类，指定参数访问地址;allowed-origins="*" 允许跨域
        // addHandler是增加处理接口和设定URL
        // addInterceptors是增加拦截器处理（可以不用）
//        registry.addHandler(myHandler(), "/ws").addInterceptors(myHandshake()).setAllowedOrigins("*");
//        registry.addHandler(myHandler(), "/sockjs/ws").addInterceptors(myHandshake()).withSockJS();

        registry.addHandler(myHandler, "/push/ws/json");
        registry.addHandler(myHandler, "/push/sockjs/json").addInterceptors(myHandshake()).withSockJS();
    }


    @Bean
    public MyHandshakeInterceptor myHandshake() {
        return new MyHandshakeInterceptor();
    }

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }


    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
                assert accessor != null;
                if (StompCommand.CONNECT.equals(accessor.getCommand())) {
                    Object raw = message.getHeaders().get(SimpMessageHeaderAccessor.NATIVE_HEADERS);
                    if (raw instanceof Map) {
                        String authorization = MapUtils.getString((Map) raw, "authorization"); //authorization登入token
                        log.info("连接成功->{}", authorization);
                    }
                } else if (StompCommand.SUBSCRIBE.equals(accessor.getCommand())) {
                    String simpDestination = (String) message.getHeaders().get("simpDestination");
                    log.info("subscribe->{}", simpDestination);
                } else if (StompCommand.UNSUBSCRIBE.equals(accessor.getCommand())) {
                    String simpDestination = (String) message.getHeaders().get("simpSubscriptionId");
                    log.info("unsubscribe->{}", simpDestination);
                } else if (StompCommand.DISCONNECT.equals(accessor.getCommand())) {
                    log.info("断开连接 {}", message);
                } else if (StompCommand.MESSAGE.equals(accessor.getCommand())) {
                    log.info("消息 {}", message);
                }  else if (StompCommand.SEND.equals(accessor.getCommand())) {
                    log.info("send {}", message);
                    messagingTemplate.convertAndSendToUser("userId3", "/topic/personalMessage", "测试发给固定用户");
                } else if (StompCommand.ERROR.equals(accessor.getCommand())) {
                    log.info("错误 {}", message);
                } else if (StompCommand.ACK.equals(accessor.getCommand())) {
                    log.info("ack {}", message);
                } else if (StompCommand.NACK.equals(accessor.getCommand())) {
                    log.info("nack {}", message);
                } else if (StompCommand.BEGIN.equals(accessor.getCommand())) {
                    log.info("begin {}", message);
                } else if (StompCommand.COMMIT.equals(accessor.getCommand())) {
                    log.info("commit {}", message);
                } else if (StompCommand.ABORT.equals(accessor.getCommand())) {
                    log.info("abort {}", message);
                }
                return message;
            }
        });
    }

}
